/*
 * Copyright (c) 2007 Pentaho Corporation.  All rights reserved. 
 * This software was developed by Pentaho Corporation and is provided under the terms 
 * of the GNU Lesser General Public License, Version 2.1. You may not use 
 * this file except in compliance with the license. If you need a copy of the license, 
 * please go to http://www.gnu.org/licenses/lgpl-2.1.txt. The Original Code is Pentaho 
 * Data Integration.  The Initial Developer is Samatar HASSAN.
 *
 * Software distributed under the GNU Lesser Public License is distributed on an "AS IS" 
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or  implied. Please refer to 
 * the license for the specific language governing your rights and limitations.
*/
package org.pentaho.di.trans.steps.stringoperations;

import org.pentaho.di.core.Const;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleStepException;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

/**
 * Apply certain operations too string.
 * 
 * @author Samatar Hassan
 * @since 02 April 2009
 */
public class StringOperations extends BaseStep implements StepInterface {
	private static Class<?> PKG = StringOperationsMeta.class; // for i18n purposes, needed by Translator2!!   $NON-NLS-1$

	private StringOperationsMeta meta;

	private StringOperationsData data;

	public StringOperations(StepMeta stepMeta, StepDataInterface stepDataInterface,
			int copyNr, TransMeta transMeta, Trans trans) {
		super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
	}

	private String processString(String string, int trimType, int lowerUpper,
			int padType, String padChar, int padLen,int iniCap, int maskHTML, 
			int digits, int removeSpecialCharacters) {
		String rcode = string;
		
		// Trim ?
		if(!Const.isEmpty(rcode)) {
			switch (trimType) {
			case StringOperationsMeta.TRIM_RIGHT:
				rcode = Const.rtrim(rcode);
				break;
			case StringOperationsMeta.TRIM_LEFT:
				rcode = Const.ltrim(rcode);
				break;
			case StringOperationsMeta.TRIM_BOTH:
				rcode = Const.trim(rcode);
				break;
			default:
				break;
			}
		}
		// Lower/Upper ?
		if(!Const.isEmpty(rcode)){
			switch (lowerUpper) {
			case StringOperationsMeta.LOWER_UPPER_LOWER:
				rcode = rcode.toLowerCase();
				break;
			case StringOperationsMeta.LOWER_UPPER_UPPER:
				rcode = rcode.toUpperCase();
				break;
			default:
				break;
			}
		}

	
		// pad String?
		if(!Const.isEmpty(rcode)) {
			switch (padType) {
			case StringOperationsMeta.PADDING_LEFT:
				rcode = Const.Lpad(rcode , padChar, padLen);
				break;
			case StringOperationsMeta.PADDING_RIGHT:
				rcode = Const.Rpad(rcode , padChar, padLen);
				break;
			default:
				break;
			}
		}
		
		// InitCap ?
		if(!Const.isEmpty(rcode)) {
			switch (iniCap) {
			case StringOperationsMeta.INIT_CAP_NO:
				break;
			case StringOperationsMeta.INIT_CAP_YES:
				rcode = Const.initCap(rcode);
				break;
			default:
				break;
			}
		}
		
		// escape ?
		if(!Const.isEmpty(rcode)) {
			switch (maskHTML) {
			case StringOperationsMeta.MASK_ESCAPE_XML:
				rcode = Const.escapeXml(rcode);
				break;
			case StringOperationsMeta.MASK_CDATA:
				rcode = Const.protectXMLCDATA(rcode);
				break;
			case StringOperationsMeta.MASK_UNESCAPE_XML:
				rcode = Const.unEscapeXml(rcode);
				break;
			case StringOperationsMeta.MASK_ESCAPE_HTML:
				rcode = Const.escapeHtml(rcode);
				break;
			case StringOperationsMeta.MASK_UNESCAPE_HTML:
				rcode = Const.unEscapeHtml(rcode);
				break;
			case StringOperationsMeta.MASK_ESCAPE_SQL:
				rcode = Const.escapeSQL(rcode);
				break;
			default:
				break;
			}
		}
		// digits only or remove digits ?
		if(!Const.isEmpty(rcode)) {
			switch (digits) {
			case StringOperationsMeta.DIGITS_NONE:
				break;
			case StringOperationsMeta.DIGITS_ONLY:
				rcode = Const.getDigitsOnly(rcode);
				break;
			case StringOperationsMeta.DIGITS_REMOVE:
				rcode = Const.removeDigits(rcode);
				break;
			default:
				break;
			}
		}
		
		// remove special characters ?
		if(!Const.isEmpty(rcode)) {
			switch (removeSpecialCharacters) {
			case StringOperationsMeta.REMOVE_SPECIAL_CHARACTERS_NONE:
				break;
			case StringOperationsMeta.REMOVE_SPECIAL_CHARACTERS_CR:
				rcode = Const.removeCR(rcode);
				break;
			case StringOperationsMeta.REMOVE_SPECIAL_CHARACTERS_LF:
				rcode = Const.removeLF(rcode);
				break;
			case StringOperationsMeta.REMOVE_SPECIAL_CHARACTERS_CRLF:
				rcode = Const.removeCRLF(rcode);
				break;
			case StringOperationsMeta.REMOVE_SPECIAL_CHARACTERS_TAB:
				rcode = Const.removeTAB(rcode);
				break;
			case StringOperationsMeta.REMOVE_SPECIAL_CHARACTERS_ESPACE:
				rcode = rcode.replace(" ", "");
				break;
			default:
				break;
			}
		}
		
		return rcode;
	}


	private synchronized Object[] processRow(RowMetaInterface rowMeta, Object[] row) throws KettleException {

		Object[] RowData = new Object[data.outputRowMeta.size()];
		// Copy the input fields.
		System.arraycopy(row, 0, RowData, 0, rowMeta.size());

		for (int i = 0; i < data.nrFieldsInStream; i++) {
			if (data.inStreamNrs[i] >= 0) {
				// Get source value
				String value = getInputRowMeta().getString(row, data.inStreamNrs[i]);
				// Apply String operations and return result value
				value= processString(value,	
						data.trimOperators[i], data.lowerUpperOperators[i],
						data.padType[i],data.padChar[i],data.padLen[i],
						data.initCap[i],data.maskHTML[i],data.digits[i],data.removeSpecialCharacters[i]);
				if(Const.isEmpty(data.outStreamNrs[i])) {
					// Update field
					RowData[data.inStreamNrs[i]]=value;
				} else {
					// create a new Field
					RowData[data.inputFieldsNr+i]=value;
				}	
			}
		}
		return RowData;
	}

	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi)
			throws KettleException {
		meta = (StringOperationsMeta) smi;
		data = (StringOperationsData) sdi;

		Object[] r = getRow(); // Get row from input rowset & set row busy!
		if (r == null)  {
			// no more input to be expected...
			setOutputDone();
			return false;
		}

		if (first) {
			first = false;
			
			// What's the format of the output row?
			data.outputRowMeta = getInputRowMeta().clone();
			data.inputFieldsNr=data.outputRowMeta.size();
			meta.getFields(data.outputRowMeta, getStepname(), null, null, this);
			data.nrFieldsInStream=meta.getFieldInStream().length;
			data.inStreamNrs = new int[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.inStreamNrs[i] = getInputRowMeta().indexOfValue(meta.getFieldInStream()[i]);
				if (data.inStreamNrs[i] < 0) // couldn't find field!
				{
					throw new KettleStepException(BaseMessages.getString(PKG, "StringOperations.Exception.FieldRequired", meta.getFieldInStream()[i])); //$NON-NLS-1$ //$NON-NLS-2$
				}
				// check field type
				if(!getInputRowMeta().getValueMeta(data.inStreamNrs[i]).isString())
					throw new KettleStepException(BaseMessages.getString(PKG, "StringOperations.Exception.FieldTypeNotString", meta.getFieldInStream()[i]));
			}
			
			data.outStreamNrs = new String[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.outStreamNrs[i] = meta.getFieldOutStream()[i];
			}
			
			// Keep track of the trim operators locally for a very small
			// optimization.
			data.trimOperators = new int[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.trimOperators[i] = meta.getTrimType()[i];
			}
			// lower Upper
			data.lowerUpperOperators = new int[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.lowerUpperOperators[i] = meta.getLowerUpper()[i];
			}
			
			// padding type?
			data.padType = new int[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.padType[i] = meta.getPaddingType()[i];
			}
			
			// padding char
			data.padChar = new String[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.padChar[i] = environmentSubstitute(meta.getPadChar()[i]);
			}
			
			// padding len
			data.padLen = new int[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.padLen[i] = Const.toInt(environmentSubstitute(meta.getPadLen()[i]), 0);
			}
			// InitCap?
			data.initCap = new int[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.initCap[i] = meta.getInitCap()[i];
			}
			// MaskXML?
			data.maskHTML = new int[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.maskHTML[i] = meta.getMaskXML()[i];
			}
			// digits?
			data.digits = new int[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.digits[i] = meta.getDigits()[i];
			}
			// remove special characters?
			data.removeSpecialCharacters = new int[data.nrFieldsInStream];
			for (int i = 0; i < meta.getFieldInStream().length; i++) {
				data.removeSpecialCharacters[i] = meta.getRemoveSpecialCharacters()[i];
			}
			
		} // end if first

        
		try {
			Object[] output = processRow(getInputRowMeta(), r);
			
			putRow(data.outputRowMeta, output);

			if (checkFeedback(getLinesRead())){
				if(log.isDetailed())
					logDetailed(BaseMessages.getString(PKG, "StringOperations.Log.LineNumber") + getLinesRead()); //$NON-NLS-1$
			}
		} catch (KettleException e) {
			
	        boolean sendToErrorRow=false;
	        String errorMessage = null;
	        
			if (getStepMeta().isDoingErrorHandling()) {
                  sendToErrorRow = true;
                  errorMessage = e.toString();
        	} else {
        		logError(BaseMessages.getString(PKG, "StringOperations.Log.ErrorInStep",e.getMessage())); //$NON-NLS-1$
				setErrors(1);
				stopAll();
				setOutputDone();  // signal end to receiver(s)
				return false;
        	}
        	if (sendToErrorRow){
        	   // Simply add this row to the error row
        	   putError(getInputRowMeta(), r, 1, errorMessage, null, "StringOperations001");
        	}
		}
		return true;
	}

	public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
		boolean rCode = true;

		meta = (StringOperationsMeta) smi;
		data = (StringOperationsData) sdi;

		if (super.init(smi, sdi)) {

			return rCode;
		}
		return false;
	}

	public void dispose(StepMetaInterface smi, StepDataInterface sdi) {
		meta = (StringOperationsMeta) smi;
		data = (StringOperationsData) sdi;

		super.dispose(smi, sdi);
	}

	public String toString() {
		return this.getClass().getName();
	}
}